media_pp\elements\filter/
tee.rs

1use std::sync::{
2    Arc, Mutex, MutexGuard, Weak,
3    atomic::{AtomicBool, AtomicU64, Ordering},
4};
5
6use crate::pp_log::{PpLog, pp_info};
7
8use crate::{
9    buffer::MediaBuffer,
10    bus::BusEvent,
11    control::ControlMsg,
12    element::{Context, Element, ElementType, Sink, element_pp_log},
13    error::Result,
14    graph::{BranchId, ElementId, GraphError, PlannedEdge, PortRef, log_topology},
15    pad::SrcPad,
16    pipeline::{ChainBuilder, DetachedBranch},
17};
18
19/// Fans a single input out to multiple sinks. [`TeeBuilder`] owns the
20/// initial fan-out; later branches are added and removed through a
21/// [`TeeHandle`], which can be cloned and used from any thread, independent
22/// of whatever thread is driving `Tee::consume`
23/// (the pipeline's source/queue-worker thread). That's the whole reason
24/// `Tee` doesn't implement [`crate::element::Source`] like other
25/// multi-pad elements (e.g. [`crate::elements::FileDemuxer`]): its pads
26/// live in individually locked branch slots instead of being a plain
27/// `&mut [SrcPad]`. `consume` only holds the branch-list lock long enough
28/// to take a cheap `Arc` snapshot, so a slow downstream does not block
29/// unrelated attach/detach operations. Detach prevents any push that has
30/// not started yet; one already executing downstream call may finish.
31///
32/// Cheap to fan out: `MediaBuffer` wraps its payload in an `Arc`, so
33/// cloning a buffer for each output is a refcount bump, not a copy of the
34/// encoded/decoded data.
35pub struct Tee {
36    pp_log: PpLog,
37    id: ElementId,
38    name: Arc<str>,
39    shared: Arc<TeeShared>,
40}
41
42struct TeeShared {
43    branches: Mutex<Vec<Arc<TeeBranch>>>,
44    next_pad_id: AtomicU64,
45    context: Arc<Context>,
46}
47
48struct TeeBranch {
49    id: Option<BranchId>,
50    root_id: ElementId,
51    active: AtomicBool,
52    pad: Mutex<SrcPad>,
53}
54
55/// Recovers the protected value after a panic instead of turning one
56/// poisoned Tee lock into a permanent source of follow-up panics. The
57/// original panic still unwinds normally; this only lets a caller that
58/// catches it keep using or detach the remaining branch state.
59fn lock_unpoisoned<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
60    match mutex.lock() {
61        Ok(guard) => guard,
62        Err(poisoned) => poisoned.into_inner(),
63    }
64}
65
66/// Build-time configuration for a [`Tee`]. Initial branches are merged
67/// with the Tee into one detached subgraph and committed by a single
68/// [`Context::attach`] call. Use [`TeeBuilder::build`] for a fixed fan-out,
69/// or [`TeeBuilder::build_dynamic`] when runtime changes need a
70/// [`TeeHandle`].
71pub struct TeeBuilder {
72    tee: Tee,
73    handle: TeeHandle,
74    initial_branches: Vec<DetachedBranch>,
75}
76
77/// A cheaply-cloneable handle for adding or removing a [`Tee`]'s sinks
78/// while the pipeline is running. It deliberately keeps only a [`Weak`]
79/// reference to the `Tee`'s shared state: retaining a handle after the
80/// pipeline finishes must not keep downstream sinks or the pipeline's
81/// [`crate::bus::Bus`] sender alive. Once the `Tee` is gone,
82/// [`TeeHandle::branch`] returns `None`.
83#[derive(Clone)]
84pub struct TeeHandle {
85    id: ElementId,
86    name: Arc<str>,
87    /// The same identity the `Tee` element logs under, cloned once at
88    /// construction — attach/detach must not rebuild it per call, and the
89    /// handle must not be able to disagree with the element about it.
90    pp_log: PpLog,
91    shared: Weak<TeeShared>,
92}
93
94impl Tee {
95    fn new(name: impl Into<String>, context: Arc<Context>) -> (Self, TeeHandle) {
96        let name: Arc<str> = name.into().into();
97        let pp_log = element_pp_log(ElementType::Tee, &name, Some(&context.pipeline_id));
98        pp_info!(pp_log: &pp_log, "created");
99        let id = context.graph.reserve_element_id();
100        let shared = Arc::new(TeeShared {
101            branches: Mutex::new(Vec::new()),
102            next_pad_id: AtomicU64::new(0),
103            context,
104        });
105        (
106            Self {
107                id,
108                name: name.clone(),
109                pp_log: pp_log.clone(),
110                shared: shared.clone(),
111            },
112            TeeHandle {
113                id,
114                name,
115                pp_log,
116                shared: Arc::downgrade(&shared),
117            },
118        )
119    }
120
121    /// Posts a branch's `push` failure to the bus under *that branch's*
122    /// own identity (via [`SrcPad::peer_identity`]) — unlike `Queue`,
123    /// which only ever has one downstream and so can only attribute a
124    /// failure to itself, `Tee` fans out to several and does know which
125    /// one just failed. Reporting it that way (rather than folding every
126    /// branch's failures into one generic `Tee` event) is what lets a
127    /// caller watching the bus tell branches apart and identify the
128    /// corresponding runtime branch. The peer identity is captured just
129    /// before the downstream call, so the event stays attributable even
130    /// if that branch is detached while the call is running.
131    fn report_branch_error(
132        &self,
133        root_id: ElementId,
134        peer: Option<(ElementType, Arc<str>)>,
135        error: crate::error::Error,
136    ) {
137        let (element_type, name) = peer.unwrap_or((ElementType::Tee, self.name.clone()));
138        self.shared.context.bus.for_element(root_id).post(
139            &self.pp_log,
140            BusEvent::Error {
141                element_type,
142                name,
143                error,
144            },
145        );
146    }
147}
148
149impl TeeShared {
150    fn next_pad(&self, tee_name: &str) -> SrcPad {
151        let id = self.next_pad_id.fetch_add(1, Ordering::Relaxed);
152        SrcPad::new(format!("{tee_name}_src{id}"))
153    }
154}
155
156impl TeeBuilder {
157    /// Starts an initially empty Tee in the supplied pipeline context.
158    pub fn new(name: impl Into<String>, context: Arc<Context>) -> Self {
159        let (tee, handle) = Tee::new(name, context);
160        Self {
161            tee,
162            handle,
163            initial_branches: Vec::new(),
164        }
165    }
166
167    /// Adds one fixed initial branch to this fan-out subgraph.
168    pub fn branch(mut self, branch: DetachedBranch) -> Self {
169        self.initial_branches.push(branch);
170        self
171    }
172
173    /// Returns the complete fixed fan-out without exposing runtime control.
174    pub fn build(self) -> Result<DetachedBranch> {
175        self.finish().map(|(branch, _handle)| branch)
176    }
177
178    /// Returns the initial subgraph together with its runtime control handle.
179    /// Attach the branch through [`Context::attach`] before using the handle.
180    pub fn build_dynamic(self) -> Result<(DetachedBranch, TeeHandle)> {
181        self.finish()
182    }
183
184    fn finish(self) -> Result<(DetachedBranch, TeeHandle)> {
185        let Self {
186            tee,
187            handle,
188            initial_branches,
189        } = self;
190        let tee_id = tee.id;
191        let tee_name = tee.name.clone();
192        let shared = tee.shared.clone();
193        let context = shared.context.clone();
194        let mut tee_branch = context.branch().to(Box::new(tee))?;
195        let mut runtime_branches = lock_unpoisoned(&shared.branches);
196
197        for branch in initial_branches {
198            let mut pad = shared.next_pad(&tee_name);
199            let from_port: Arc<str> = pad.name().into();
200            let DetachedBranch { root, plan } = branch;
201            let root_id = plan.root;
202
203            tee_branch.plan.edges.push(PlannedEdge {
204                from: PortRef {
205                    element: tee_id,
206                    port: from_port,
207                },
208                to: PortRef {
209                    element: root_id,
210                    port: "sink".into(),
211                },
212            });
213            tee_branch.plan.nodes.extend(plan.nodes);
214            tee_branch.plan.edges.extend(plan.edges);
215
216            pad.link(root);
217            runtime_branches.push(Arc::new(TeeBranch {
218                id: None,
219                root_id,
220                active: AtomicBool::new(true),
221                pad: Mutex::new(pad),
222            }));
223        }
224        drop(runtime_branches);
225
226        Ok((tee_branch, handle))
227    }
228}
229
230impl TeeHandle {
231    /// A [`crate::pipeline::ChainBuilder`] pre-wired with this `Tee`'s own
232    /// [`Context`] — lets a caller build a whole new branch (`.pipe(...)`
233    /// chains, ending in `.to(...)`) at any point after the pipeline
234    /// started running, then hand the result to [`TeeHandle::attach`],
235    /// without needing to retain the pipeline context separately
236    /// around separately. Returns `None` once the `Tee` has been dropped.
237    pub fn branch(&self) -> Option<ChainBuilder> {
238        let shared = self.shared.upgrade()?;
239        Some(shared.context.branch())
240    }
241
242    /// Attaches a runtime branch, returning the stable ID used to remove it.
243    /// Fixed initial branches belong in [`TeeBuilder`].
244    pub fn attach(&self, branch: DetachedBranch) -> Result<BranchId> {
245        let shared = self
246            .shared
247            .upgrade()
248            .ok_or(GraphError::ParentNotAttached(self.id))?;
249        let mut branches = lock_unpoisoned(&shared.branches);
250        let mut pad = shared.next_pad(&self.name);
251        let from_port: Arc<str> = pad.name().into();
252        let DetachedBranch { root, plan } = branch;
253        let root_id = plan.root;
254        let branch_id =
255            shared
256                .context
257                .graph
258                .attach_with(self.id, from_port, plan, |branch_id| {
259                    pad.link(root);
260                    branches.push(Arc::new(TeeBranch {
261                        id: Some(branch_id),
262                        root_id,
263                        active: AtomicBool::new(true),
264                        pad: Mutex::new(pad),
265                    }));
266                    Ok(())
267                })?;
268        let snapshot =
269            crate::log::enabled(crate::log::Level::Info).then(|| shared.context.graph.snapshot());
270        drop(branches);
271        if let Some(snapshot) = snapshot {
272            log_topology(&self.pp_log, "attach", &snapshot);
273        }
274        Ok(branch_id)
275    }
276
277    /// Detaches exactly the branch returned by [`TeeHandle::attach`]. The
278    /// runtime peer and every graph node owned by it disappear in the same
279    /// transaction. Names are deliberately not used as graph keys.
280    pub fn detach(&self, branch_id: BranchId) -> Result<()> {
281        let shared = self
282            .shared
283            .upgrade()
284            .ok_or(GraphError::BranchNotAttached(branch_id))?;
285        let mut branches = lock_unpoisoned(&shared.branches);
286        let index = branches
287            .iter()
288            .position(|branch| branch.id == Some(branch_id))
289            .ok_or(GraphError::BranchNotAttached(branch_id))?;
290        let mut removed = None;
291        shared.context.graph.detach_with(branch_id, || {
292            let branch = branches.remove(index);
293            branch.active.store(false, Ordering::Release);
294            removed = Some(branch);
295            Ok(())
296        })?;
297        let snapshot =
298            crate::log::enabled(crate::log::Level::Info).then(|| shared.context.graph.snapshot());
299        drop(branches);
300        // The last Arc owns the downstream sink. Dropping it outside both
301        // the branch-list and graph locks allows arbitrary Sink::drop code
302        // to inspect the graph or call back into Tee without deadlocking.
303        drop(removed);
304        if let Some(snapshot) = snapshot {
305            log_topology(&self.pp_log, "detach", &snapshot);
306        }
307        Ok(())
308    }
309
310    /// Resolves the owning branch from any element ID inside it and
311    /// detaches that branch. Useful when an error is attributed to a stage
312    /// behind a queue rather than to the branch root.
313    pub fn detach_branch_containing(&self, element: ElementId) -> Result<()> {
314        let shared = self
315            .shared
316            .upgrade()
317            .ok_or(GraphError::ParentNotAttached(self.id))?;
318        let branch_id = shared
319            .context
320            .graph
321            .branch_containing(element)
322            .ok_or(GraphError::ParentNotAttached(element))?;
323        self.detach(branch_id)
324    }
325
326    pub fn sink_count(&self) -> usize {
327        self.shared
328            .upgrade()
329            .map(|shared| lock_unpoisoned(&shared.branches).len())
330            .unwrap_or(0)
331    }
332}
333
334impl Element for Tee {
335    fn name(&self) -> Arc<str> {
336        self.name.clone()
337    }
338
339    fn element_type(&self) -> ElementType {
340        ElementType::Tee
341    }
342
343    fn graph_id(&self) -> Option<ElementId> {
344        Some(self.id)
345    }
346
347    fn pp_log(&self) -> &PpLog {
348        &self.pp_log
349    }
350
351    fn pp_log_mut(&mut self) -> &mut PpLog {
352        &mut self.pp_log
353    }
354}
355
356impl Sink for Tee {
357    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
358        let branches = lock_unpoisoned(&self.shared.branches).clone();
359        // One branch failing must not stop the buffer from reaching its
360        // siblings — same "errors never kill anything, just get reported"
361        // rule `Queue`'s worker loop follows. That buffer is dropped for
362        // the failing branch only; the branch itself stays wired and gets
363        // retried on the next one. Whoever's watching the bus decides
364        // whether to call `TeeHandle::detach` for it.
365        for branch in branches {
366            if !branch.active.load(Ordering::Acquire) {
367                continue;
368            }
369            let mut pad = lock_unpoisoned(&branch.pad);
370            // Detach may have won the race while this thread waited for a
371            // previous push on the same branch to finish.
372            if !branch.active.load(Ordering::Acquire) {
373                continue;
374            }
375            let peer = pad.peer_identity();
376            let outcome = pad.push(buf.clone());
377            drop(pad);
378            if let Err(error) = outcome {
379                self.report_branch_error(branch.root_id, peer, error);
380            }
381        }
382        Ok(())
383    }
384
385    fn control(&mut self, msg: ControlMsg) -> Result<()> {
386        // Control failures follow the same isolation rule as data failures:
387        // report the failed branch, but still deliver the message to every
388        // sibling. This is especially important for Stop and Pause.
389        let branches = lock_unpoisoned(&self.shared.branches).clone();
390        for branch in branches {
391            if !branch.active.load(Ordering::Acquire) {
392                continue;
393            }
394            let mut pad = lock_unpoisoned(&branch.pad);
395            if !branch.active.load(Ordering::Acquire) {
396                continue;
397            }
398            let peer = pad.peer_identity();
399            let outcome = pad.control(msg);
400            drop(pad);
401            if let Err(error) = outcome {
402                self.report_branch_error(branch.root_id, peer, error);
403            }
404        }
405        Ok(())
406    }
407}
408
409#[cfg(test)]
410mod tests {
411    use std::{
412        panic::{AssertUnwindSafe, catch_unwind},
413        sync::{
414            Barrier,
415            atomic::{AtomicBool, AtomicUsize, Ordering},
416            mpsc,
417        },
418        thread,
419        time::Duration,
420    };
421
422    use super::*;
423    use crate::{bus::Bus, graph::PipelineGraph};
424
425    fn packet() -> MediaBuffer {
426        MediaBuffer::Packet(Arc::new(ffmpeg_next::Packet::empty()))
427    }
428
429    struct CountingSink {
430        pp_log: PpLog,
431        name: &'static str,
432        count: Arc<AtomicUsize>,
433    }
434
435    impl Element for CountingSink {
436        fn name(&self) -> Arc<str> {
437            self.name.into()
438        }
439
440        fn element_type(&self) -> ElementType {
441            ElementType::Other
442        }
443
444        fn pp_log(&self) -> &PpLog {
445            &self.pp_log
446        }
447
448        fn pp_log_mut(&mut self) -> &mut PpLog {
449            &mut self.pp_log
450        }
451    }
452
453    impl Sink for CountingSink {
454        fn consume(&mut self, _buf: MediaBuffer) -> Result<()> {
455            self.count.fetch_add(1, Ordering::SeqCst);
456            Ok(())
457        }
458
459        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
460            Ok(())
461        }
462    }
463
464    struct AlwaysFailSink {
465        pp_log: PpLog,
466    }
467
468    impl Element for AlwaysFailSink {
469        fn name(&self) -> Arc<str> {
470            "always-fail".into()
471        }
472
473        fn element_type(&self) -> ElementType {
474            ElementType::Other
475        }
476
477        fn pp_log(&self) -> &PpLog {
478            &self.pp_log
479        }
480
481        fn pp_log_mut(&mut self) -> &mut PpLog {
482            &mut self.pp_log
483        }
484    }
485
486    impl Sink for AlwaysFailSink {
487        fn consume(&mut self, _buf: MediaBuffer) -> Result<()> {
488            Err(crate::error::Error::Other(
489                "simulated branch failure".into(),
490            ))
491        }
492
493        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
494            Ok(())
495        }
496    }
497
498    struct ControlObservingSink {
499        pp_log: PpLog,
500        name: &'static str,
501        count: Arc<AtomicUsize>,
502        fail: bool,
503    }
504
505    impl Element for ControlObservingSink {
506        fn name(&self) -> Arc<str> {
507            self.name.into()
508        }
509
510        fn element_type(&self) -> ElementType {
511            ElementType::Other
512        }
513
514        fn pp_log(&self) -> &PpLog {
515            &self.pp_log
516        }
517
518        fn pp_log_mut(&mut self) -> &mut PpLog {
519            &mut self.pp_log
520        }
521    }
522
523    impl Sink for ControlObservingSink {
524        fn consume(&mut self, _buf: MediaBuffer) -> Result<()> {
525            Ok(())
526        }
527
528        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
529            self.count.fetch_add(1, Ordering::SeqCst);
530            if self.fail {
531                Err(crate::error::Error::Other(
532                    "simulated control failure".into(),
533                ))
534            } else {
535                Ok(())
536            }
537        }
538    }
539
540    struct PanicOnceSink {
541        pp_log: PpLog,
542        panicked: bool,
543        successful: Arc<AtomicUsize>,
544    }
545
546    impl Element for PanicOnceSink {
547        fn name(&self) -> Arc<str> {
548            "panic-once".into()
549        }
550
551        fn element_type(&self) -> ElementType {
552            ElementType::Other
553        }
554
555        fn pp_log(&self) -> &PpLog {
556            &self.pp_log
557        }
558
559        fn pp_log_mut(&mut self) -> &mut PpLog {
560            &mut self.pp_log
561        }
562    }
563
564    impl Sink for PanicOnceSink {
565        fn consume(&mut self, _buf: MediaBuffer) -> Result<()> {
566            if !self.panicked {
567                self.panicked = true;
568                panic!("simulated downstream panic");
569            }
570            self.successful.fetch_add(1, Ordering::SeqCst);
571            Ok(())
572        }
573
574        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
575            Ok(())
576        }
577    }
578
579    struct BlockingSink {
580        pp_log: PpLog,
581        entered: Option<mpsc::Sender<()>>,
582        release: mpsc::Receiver<()>,
583    }
584
585    impl Element for BlockingSink {
586        fn name(&self) -> Arc<str> {
587            "blocking".into()
588        }
589
590        fn element_type(&self) -> ElementType {
591            ElementType::Other
592        }
593
594        fn pp_log(&self) -> &PpLog {
595            &self.pp_log
596        }
597
598        fn pp_log_mut(&mut self) -> &mut PpLog {
599            &mut self.pp_log
600        }
601    }
602
603    impl Sink for BlockingSink {
604        fn consume(&mut self, _buf: MediaBuffer) -> Result<()> {
605            if let Some(entered) = self.entered.take() {
606                let _ = entered.send(());
607            }
608            let _ = self.release.recv();
609            Ok(())
610        }
611
612        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
613            Ok(())
614        }
615    }
616
617    struct GraphInspectingDropSink {
618        pp_log: PpLog,
619        graph: PipelineGraph,
620        dropped: Option<mpsc::Sender<()>>,
621    }
622
623    impl Element for GraphInspectingDropSink {
624        fn name(&self) -> Arc<str> {
625            "graph-inspecting-drop".into()
626        }
627
628        fn element_type(&self) -> ElementType {
629            ElementType::Other
630        }
631
632        fn pp_log(&self) -> &PpLog {
633            &self.pp_log
634        }
635
636        fn pp_log_mut(&mut self) -> &mut PpLog {
637            &mut self.pp_log
638        }
639    }
640
641    impl Sink for GraphInspectingDropSink {
642        fn consume(&mut self, _buf: MediaBuffer) -> Result<()> {
643            Ok(())
644        }
645
646        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
647            Ok(())
648        }
649    }
650
651    impl Drop for GraphInspectingDropSink {
652        fn drop(&mut self) {
653            let _ = self.graph.snapshot();
654            if let Some(dropped) = self.dropped.take() {
655                let _ = dropped.send(());
656            }
657        }
658    }
659
660    /// A failing branch must not stop the same buffer from reaching its
661    /// siblings, nor stop `Tee::consume` itself from returning `Ok` — same
662    /// "errors get reported, nothing dies" rule `Queue`'s worker loop
663    /// follows (see `queue::tests::a_failing_consume_drops_that_buffer_but_keeps_the_worker_alive`).
664    /// Wires the failing branch in the *middle* (`before`/`after` on
665    /// either side of it) so the test also proves a mid-`rest` failure
666    /// doesn't short-circuit the fan-out to what comes after it,
667    /// including `last`.
668    #[test]
669    fn a_failing_branch_does_not_block_its_siblings() {
670        let (bus, bus_rx) = Bus::new();
671        let graph = PipelineGraph::new();
672        let source_id = graph.add_source(ElementType::Other, "source".into());
673        let context = Arc::new(Context::for_test(bus, "test", graph, source_id));
674        let before_count = Arc::new(AtomicUsize::new(0));
675        let after_count = Arc::new(AtomicUsize::new(0));
676        let before = context
677            .branch()
678            .to(Box::new(CountingSink {
679                name: "before",
680                count: before_count.clone(),
681                pp_log: element_pp_log(ElementType::Other, "before", None),
682            }))
683            .unwrap();
684        let failing = context
685            .branch()
686            .to(Box::new(AlwaysFailSink {
687                pp_log: element_pp_log(ElementType::Other, "always-fail", None),
688            }))
689            .unwrap();
690        let after = context
691            .branch()
692            .to(Box::new(CountingSink {
693                name: "after",
694                count: after_count.clone(),
695                pp_log: element_pp_log(ElementType::Other, "after", None),
696            }))
697            .unwrap();
698        let tee_branch = TeeBuilder::new("tee", context.clone())
699            .branch(before)
700            .branch(failing)
701            .branch(after)
702            .build()
703            .unwrap();
704        let mut upstream = SrcPad::new("source_src");
705        context.attach_pad(&mut upstream, tee_branch).unwrap();
706
707        for _ in 0..3 {
708            upstream
709                .push(packet())
710                .expect("a branch failing must not surface as an error from Tee::consume");
711        }
712
713        assert_eq!(before_count.load(Ordering::SeqCst), 3);
714        assert_eq!(after_count.load(Ordering::SeqCst), 3);
715
716        drop(upstream);
717        drop(context);
718        let errors: Vec<_> = bus_rx
719            .iter()
720            .filter(|e| matches!(e, BusEvent::Error { .. }))
721            .collect();
722        assert_eq!(
723            errors.len(),
724            3,
725            "expected one Error event per failed push, not a fatal short-circuit"
726        );
727        assert!(
728            errors.iter().all(|e| matches!(
729                e,
730                BusEvent::Error { name, .. } if &**name == "always-fail"
731            )),
732            "each Error event should be attributed to the branch that actually \
733             failed, not to Tee itself — that's what lets a caller call \
734             TeeHandle::detach(branch_id) straight off the bus; got {errors:?}"
735        );
736    }
737
738    #[test]
739    fn a_failing_control_branch_does_not_block_its_siblings() {
740        let (bus, bus_rx) = Bus::new();
741        let graph = PipelineGraph::new();
742        let source_id = graph.add_source(ElementType::Other, "source".into());
743        let context = Arc::new(Context::for_test(bus, "test", graph, source_id));
744        let failing_count = Arc::new(AtomicUsize::new(0));
745        let healthy_count = Arc::new(AtomicUsize::new(0));
746        let failing = context
747            .branch()
748            .to(Box::new(ControlObservingSink {
749                name: "control-fail",
750                count: failing_count.clone(),
751                fail: true,
752                pp_log: element_pp_log(ElementType::Other, "control-fail", None),
753            }))
754            .unwrap();
755        let healthy = context
756            .branch()
757            .to(Box::new(ControlObservingSink {
758                name: "control-ok",
759                count: healthy_count.clone(),
760                fail: false,
761                pp_log: element_pp_log(ElementType::Other, "control-ok", None),
762            }))
763            .unwrap();
764        let tee_branch = TeeBuilder::new("tee", context.clone())
765            .branch(failing)
766            .branch(healthy)
767            .build()
768            .unwrap();
769        let mut upstream = SrcPad::new("source_src");
770        context.attach_pad(&mut upstream, tee_branch).unwrap();
771
772        upstream
773            .control(ControlMsg::Pause)
774            .expect("a branch control failure should be reported, not short-circuit Tee");
775
776        assert_eq!(failing_count.load(Ordering::SeqCst), 1);
777        assert_eq!(healthy_count.load(Ordering::SeqCst), 1);
778        let message = bus_rx
779            .try_recv_message()
780            .expect("the failing control branch should post an Error event");
781        assert!(matches!(
782            message.event,
783            BusEvent::Error { name, .. } if &*name == "control-fail"
784        ));
785        assert!(bus_rx.try_recv_message().is_none());
786    }
787
788    #[test]
789    fn a_poisoned_branch_pad_can_be_used_after_the_panic_is_caught() {
790        let (bus, _bus_rx) = Bus::new();
791        let graph = PipelineGraph::new();
792        let source_id = graph.add_source(ElementType::Other, "source".into());
793        let context = Arc::new(Context::for_test(bus, "test", graph, source_id));
794        let successful = Arc::new(AtomicUsize::new(0));
795        let panic_once = context
796            .branch()
797            .to(Box::new(PanicOnceSink {
798                panicked: false,
799                successful: successful.clone(),
800                pp_log: element_pp_log(ElementType::Other, "panic-once", None),
801            }))
802            .unwrap();
803        let tee_branch = TeeBuilder::new("tee", context.clone())
804            .branch(panic_once)
805            .build()
806            .unwrap();
807        let mut upstream = SrcPad::new("source_src");
808        context.attach_pad(&mut upstream, tee_branch).unwrap();
809
810        let first = catch_unwind(AssertUnwindSafe(|| upstream.push(packet())));
811        assert!(
812            first.is_err(),
813            "the original downstream panic must propagate"
814        );
815        upstream
816            .push(packet())
817            .expect("the poisoned branch pad should be recovered on the next push");
818        assert_eq!(successful.load(Ordering::SeqCst), 1);
819    }
820
821    #[test]
822    fn a_poisoned_branch_list_does_not_break_attach_or_detach() {
823        let (bus, _bus_rx) = Bus::new();
824        let graph = PipelineGraph::new();
825        let source_id = graph.add_source(ElementType::Other, "source".into());
826        let context = Arc::new(Context::for_test(bus, "test", graph, source_id));
827        let (tee_branch, handle) = TeeBuilder::new("tee", context.clone())
828            .build_dynamic()
829            .unwrap();
830        let mut upstream = SrcPad::new("source_src");
831        context.attach_pad(&mut upstream, tee_branch).unwrap();
832        let shared = handle.shared.upgrade().unwrap();
833
834        let poisoned = catch_unwind(AssertUnwindSafe(|| {
835            let _branches = shared.branches.lock().unwrap();
836            panic!("poison the branch-list lock");
837        }));
838        assert!(poisoned.is_err());
839        assert_eq!(handle.sink_count(), 0);
840
841        let branch = handle
842            .branch()
843            .unwrap()
844            .to(Box::new(CountingSink {
845                name: "after-poison",
846                count: Arc::new(AtomicUsize::new(0)),
847                pp_log: element_pp_log(ElementType::Other, "after-poison", None),
848            }))
849            .unwrap();
850        let branch_id = handle.attach(branch).unwrap();
851        assert_eq!(handle.sink_count(), 1);
852        handle.detach(branch_id).unwrap();
853        assert_eq!(handle.sink_count(), 0);
854    }
855
856    #[test]
857    fn blocked_downstream_does_not_block_unrelated_attach_or_detach() {
858        let (bus, _bus_rx) = Bus::new();
859        let graph = PipelineGraph::new();
860        let source_id = graph.add_source(ElementType::Other, "source".into());
861        let context = Arc::new(Context::for_test(bus, "test", graph, source_id));
862        let (tee_branch, handle) = TeeBuilder::new("tee", context.clone())
863            .build_dynamic()
864            .unwrap();
865        let mut upstream = SrcPad::new("source_src");
866        context.attach_pad(&mut upstream, tee_branch).unwrap();
867
868        let (entered_tx, entered_rx) = mpsc::channel();
869        let (release_tx, release_rx) = mpsc::channel();
870        let blocking = handle
871            .branch()
872            .unwrap()
873            .to(Box::new(BlockingSink {
874                entered: Some(entered_tx),
875                release: release_rx,
876                pp_log: element_pp_log(ElementType::Other, "blocking", None),
877            }))
878            .unwrap();
879        let blocking_id = handle.attach(blocking).unwrap();
880
881        let push_thread = thread::spawn(move || upstream.push(packet()));
882        entered_rx
883            .recv_timeout(Duration::from_secs(1))
884            .expect("blocking branch was never entered");
885
886        let new_branch = handle
887            .branch()
888            .unwrap()
889            .to(Box::new(CountingSink {
890                name: "new",
891                count: Arc::new(AtomicUsize::new(0)),
892                pp_log: element_pp_log(ElementType::Other, "new", None),
893            }))
894            .unwrap();
895        let (attach_tx, attach_rx) = mpsc::channel();
896        let attach_handle = handle.clone();
897        let attach_thread = thread::spawn(move || {
898            let _ = attach_tx.send(attach_handle.attach(new_branch));
899        });
900
901        let (detach_tx, detach_rx) = mpsc::channel();
902        let detach_handle = handle.clone();
903        let detach_thread = thread::spawn(move || {
904            let _ = detach_tx.send(detach_handle.detach(blocking_id));
905        });
906
907        let attach_before_release = attach_rx.recv_timeout(Duration::from_millis(250)).ok();
908        let detach_before_release = detach_rx.recv_timeout(Duration::from_millis(250)).ok();
909        let attach_completed_while_blocked = attach_before_release.is_some();
910        let detach_completed_while_blocked = detach_before_release.is_some();
911        let _ = release_tx.send(());
912
913        push_thread.join().unwrap().unwrap();
914        attach_thread.join().unwrap();
915        detach_thread.join().unwrap();
916        let attach_result = attach_before_release
917            .unwrap_or_else(|| attach_rx.recv_timeout(Duration::from_secs(1)).unwrap());
918        let detach_result = detach_before_release
919            .unwrap_or_else(|| detach_rx.recv_timeout(Duration::from_secs(1)).unwrap());
920        attach_result.unwrap();
921        detach_result.unwrap();
922
923        assert!(
924            attach_completed_while_blocked,
925            "an unrelated attach waited for the blocked downstream"
926        );
927        assert!(
928            detach_completed_while_blocked,
929            "detach waited for an already-running downstream call"
930        );
931    }
932
933    #[test]
934    fn concurrent_push_attach_and_detach_stays_consistent_under_stress() {
935        const MIN_PUSHES: usize = 10_000;
936        const MUTATIONS: usize = 500;
937
938        let (bus, _bus_rx) = Bus::new();
939        let graph = PipelineGraph::new();
940        let source_id = graph.add_source(ElementType::Other, "source".into());
941        let context = Arc::new(Context::for_test(bus, "test", graph, source_id));
942        let initial_count = Arc::new(AtomicUsize::new(0));
943        let initial = context
944            .branch()
945            .to(Box::new(CountingSink {
946                name: "initial",
947                count: initial_count.clone(),
948                pp_log: element_pp_log(ElementType::Other, "initial", None),
949            }))
950            .unwrap();
951        let (tee_branch, handle) = TeeBuilder::new("tee", context.clone())
952            .branch(initial)
953            .build_dynamic()
954            .unwrap();
955        let mut upstream = SrcPad::new("source_src");
956        context.attach_pad(&mut upstream, tee_branch).unwrap();
957
958        let start = Arc::new(Barrier::new(2));
959        let mutating = Arc::new(AtomicBool::new(true));
960        let (push_done_tx, push_done_rx) = mpsc::channel();
961        let push_start = start.clone();
962        let push_mutating = mutating.clone();
963        let push_thread = thread::spawn(move || {
964            push_start.wait();
965            let packet = packet();
966            let mut pushed = 0;
967            let mut outcome = Ok(());
968            while push_mutating.load(Ordering::Acquire) || pushed < MIN_PUSHES {
969                if let Err(error) = upstream.push(packet.clone()) {
970                    outcome = Err(error.to_string());
971                    break;
972                }
973                pushed += 1;
974                if pushed % 32 == 0 {
975                    thread::yield_now();
976                }
977            }
978            let _ = push_done_tx.send((upstream, outcome, pushed));
979        });
980
981        let (mutation_done_tx, mutation_done_rx) = mpsc::channel();
982        let mutation_start = start;
983        let mutation_handle = handle.clone();
984        let mutation_thread = thread::spawn(move || {
985            mutation_start.wait();
986            let outcome = (|| -> std::result::Result<(), String> {
987                for _ in 0..MUTATIONS {
988                    let branch = mutation_handle
989                        .branch()
990                        .ok_or_else(|| "Tee disappeared during stress test".to_owned())?
991                        .to(Box::new(CountingSink {
992                            name: "dynamic",
993                            count: Arc::new(AtomicUsize::new(0)),
994                            pp_log: element_pp_log(ElementType::Other, "dynamic", None),
995                        }))
996                        .map_err(|error| error.to_string())?;
997                    let branch_id = mutation_handle
998                        .attach(branch)
999                        .map_err(|error| error.to_string())?;
1000                    thread::yield_now();
1001                    mutation_handle
1002                        .detach(branch_id)
1003                        .map_err(|error| error.to_string())?;
1004                }
1005                Ok(())
1006            })();
1007            mutating.store(false, Ordering::Release);
1008            let _ = mutation_done_tx.send(outcome);
1009        });
1010
1011        mutation_done_rx
1012            .recv_timeout(Duration::from_secs(10))
1013            .expect("attach/detach stress thread timed out")
1014            .expect("attach/detach stress thread failed");
1015        let (upstream, push_outcome, pushed) = push_done_rx
1016            .recv_timeout(Duration::from_secs(10))
1017            .expect("push stress thread timed out");
1018        push_outcome.expect("push stress thread failed");
1019        push_thread.join().unwrap();
1020        mutation_thread.join().unwrap();
1021
1022        assert!(pushed >= MIN_PUSHES);
1023        assert_eq!(initial_count.load(Ordering::SeqCst), pushed);
1024        assert_eq!(handle.sink_count(), 1);
1025        let graph = context.graph.snapshot();
1026        assert_eq!(graph.nodes.len(), 3);
1027        assert_eq!(graph.edges.len(), 2);
1028        assert_eq!(graph.revision, 2 + (MUTATIONS as u64 * 2));
1029        drop(upstream);
1030    }
1031
1032    #[test]
1033    fn detached_sink_is_dropped_outside_the_graph_lock() {
1034        let (bus, _bus_rx) = Bus::new();
1035        let graph = PipelineGraph::new();
1036        let source_id = graph.add_source(ElementType::Other, "source".into());
1037        let context = Arc::new(Context::for_test(bus, "test", graph.clone(), source_id));
1038        let (tee_branch, handle) = TeeBuilder::new("tee", context.clone())
1039            .build_dynamic()
1040            .unwrap();
1041        let mut upstream = SrcPad::new("source_src");
1042        context.attach_pad(&mut upstream, tee_branch).unwrap();
1043
1044        let (dropped_tx, dropped_rx) = mpsc::channel();
1045        let branch = handle
1046            .branch()
1047            .unwrap()
1048            .to(Box::new(GraphInspectingDropSink {
1049                graph,
1050                dropped: Some(dropped_tx),
1051                pp_log: element_pp_log(ElementType::Other, "graph-inspecting-drop", None),
1052            }))
1053            .unwrap();
1054        let branch_id = handle.attach(branch).unwrap();
1055        let (done_tx, done_rx) = mpsc::channel();
1056        let detach_thread = thread::spawn(move || {
1057            let _ = done_tx.send(handle.detach(branch_id));
1058        });
1059
1060        dropped_rx
1061            .recv_timeout(Duration::from_secs(1))
1062            .expect("sink Drop deadlocked while inspecting the graph");
1063        done_rx
1064            .recv_timeout(Duration::from_secs(1))
1065            .expect("detach did not finish")
1066            .unwrap();
1067        detach_thread.join().unwrap();
1068        drop(upstream);
1069    }
1070
1071    #[test]
1072    fn retained_handle_does_not_keep_tee_context_or_bus_alive() {
1073        let (bus, bus_rx) = Bus::new();
1074        let graph = PipelineGraph::new();
1075        let source_id = graph.add_source(ElementType::Other, "source".into());
1076        let context = Arc::new(Context::for_test(bus, "test", graph, source_id));
1077        let (tee_branch, handle) = TeeBuilder::new("tee", context.clone())
1078            .build_dynamic()
1079            .unwrap();
1080
1081        drop(context);
1082        drop(tee_branch);
1083
1084        assert!(handle.branch().is_none());
1085        assert_eq!(handle.sink_count(), 0);
1086        assert!(
1087            bus_rx.iter().next().is_none(),
1088            "a retained TeeHandle must not keep the Bus sender alive"
1089        );
1090    }
1091}
media_pp\elements\filter/tee.rs

media_pp\elements\filter/
tee.rs
